package yanglib.session;

import yanglib.message.Message;
import yanglib.network.MessageHandler;
import yanglib.network.Network;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created with IntelliJ IDEA.
 * User: user
 * Date: 9/9/14
 * Time: 1:37 PM
 * To change this template use File | Settings | File Templates.
 */
public class ClientSession implements MessageHandler{
    public static final int OUTSTANDING_QUEUE_SIZE = 100;
    private int selfID;
    private Network network;
    private boolean orderedReply;
    private long clientRequestID = 0;
    private LinkedBlockingDeque<PendingRequest> outstandingRequests = new LinkedBlockingDeque<PendingRequest>(OUTSTANDING_QUEUE_SIZE);
    private TreeMap<Long, PendingRequest> requestsWaitingForReply = new TreeMap<Long, PendingRequest>();
    private TimeoutChecker timeoutChecker = new TimeoutChecker();
    private boolean running = true;
    private SendingThread sendingThread = new SendingThread();

    public ClientSession(int selfID, boolean orderedReply){
        this.selfID = selfID;
        this.orderedReply = orderedReply;
        sendingThread.start();
        timeoutChecker.start();
    }

    public void setNetwork(Network network){
        this.network = network;
    }

    public void submitRequest(Request request, Callback callback, Set<Integer> targets, long timeout){
        request.setRequestID(clientRequestID);
        synchronized(this){
            clientRequestID ++;
        }
        PendingRequest pendingRequest = new PendingRequest(this, request, callback, targets, timeout);
        timeoutChecker.add(pendingRequest);
        try {
            outstandingRequests.put(pendingRequest);
        } catch (InterruptedException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }
    }

    protected void resend(PendingRequest pRequest, Set<Integer> targets, long timeout){
        for(Integer target: targets){
            try{
                network.send(pRequest.getRequest(), target);
                System.out.println("Resent "+pRequest.getRequest()+" to "+target+" requestID="+pRequest.getRequest().getRequestID());
                try{
                    Thread.sleep(60000);
                }
                catch(InterruptedException e){
                    e.printStackTrace();
                }
            }
            catch(IOException e){
                pRequest.getCallback().handleError(pRequest, target, e);
            }
            pRequest.getTargets().add(target);
        }
        timeoutChecker.reset(pRequest, timeout);
    }

    private synchronized void addToWaitingList(PendingRequest pRequest){
        while(requestsWaitingForReply.size() >= OUTSTANDING_QUEUE_SIZE){
            try{
                this.wait();
            }
            catch(InterruptedException e){
                e.printStackTrace();
            }
        }
        requestsWaitingForReply.put(pRequest.getRequest().getRequestID(), pRequest);
    }

    private class SendingThread extends Thread{

        public void run(){
            while(running){
                try{
                    PendingRequest pRequest = outstandingRequests.take();
                    addToWaitingList(pRequest);
                    for(Integer target: pRequest.getTargets()){
                        try{
                            network.send(pRequest.getRequest(), target);
                            //System.out.println("Sent "+pRequest.getRequest()+" to "+target+" requestID="+pRequest.getRequest().getRequestID());
                        }
                        catch(IOException e){
                            pRequest.getCallback().handleError(pRequest, target, e);
                        }
                    }

                }
                catch(InterruptedException e){
                    e.printStackTrace();
                    return;
                }
            }
        }
    }

    protected void resetTimeout(PendingRequest pRequest, long timeout){
        timeoutChecker.reset(pRequest, timeout);
    }

    public synchronized void handleMessage(Message message){
        if(!(message instanceof Reply)){
            throw new RuntimeException("Unexpected message "+message.getClass());
        }
        Reply reply = (Reply)message;
        if(!requestsWaitingForReply.containsKey(reply.getRequestID())){
            //System.out.println("Got a reply "+reply.getRequestID()+" not in outstanding, maybe an old reply");
            return;
        }
        PendingRequest pRequest = requestsWaitingForReply.get(reply.getRequestID());
        boolean completed = pRequest.getCallback().handleReply(pRequest, reply);


        if(completed){
            pRequest.setCompleted();
            //System.out.println(pRequest.getRequest().getRequestID()+" is completed");
            try{
                timeoutChecker.remove(pRequest);
            }
            catch(NullPointerException e){
                e.printStackTrace();
                System.exit(-1);
            }
            if(!this.orderedReply){
                pRequest.getCallback().complete(pRequest);
                requestsWaitingForReply.remove(reply.getRequestID());
                this.notifyAll();
            }else{
                Iterator<Map.Entry<Long, PendingRequest>> iter = requestsWaitingForReply.entrySet().iterator();
                while(iter.hasNext()){
                    Map.Entry<Long, PendingRequest> e = iter.next();
                    if(e.getValue().isCompleted()){
                        e.getValue().getCallback().complete(e.getValue());
                        iter.remove();
                    }
                    else{
                        break;
                    }
                }
                this.notifyAll();
            }
        }

    }
}
